package cn.devzyh.oms.wsms.service.impl;

import cn.devzyh.oms.common.core.message.CommonErrorCodes;
import cn.devzyh.oms.common.core.message.WsmsErrorCodes;
import cn.devzyh.oms.common.core.util.R;
import cn.devzyh.oms.common.core.util.RedisUtils;
import cn.devzyh.oms.common.rabbitmq.queue.WsmsQueues;
import cn.devzyh.oms.wsms.api.constant.StockConstants;
import cn.devzyh.oms.wsms.api.converter.StockConverter;
import cn.devzyh.oms.wsms.api.dto.StockDTO;
import cn.devzyh.oms.wsms.api.dto.StockUpdateDTO;
import cn.devzyh.oms.wsms.api.entity.Stock;
import cn.devzyh.oms.wsms.api.entity.StockLog;
import cn.devzyh.oms.wsms.api.enums.StockLocationEnum;
import cn.devzyh.oms.wsms.api.vo.StockVO;
import cn.devzyh.oms.wsms.mapper.StockMapper;
import cn.devzyh.oms.wsms.service.*;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.lock.LockInfo;
import com.baomidou.lock.LockTemplate;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.stereotype.Service;

import java.util.*;

/**
 * 基于缓存实现的库存操作
 *
 * @author devzyh
 * @date 2023-10-16
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class StockServiceImpl extends ServiceImpl<StockMapper, Stock> implements StockService {

	private final StringRedisTemplate stringRedisTemplate;

	private final LockTemplate lockTemplate;

	private final AmqpTemplate amqpTemplate;

	private final WarehouseService warehouseService;

	private final ChannelService channelService;

	private final SkuService skuService;

	private final StockLogService logService;

	/**
	 * 库存操作LUA脚本
	 */
	private static final RedisScript<String> STOCK_SCRIPT = RedisScript.of("""
			local stock = redis.call('HMGET', KEYS[1], 'quantity', 'version')
			local qty = stock[1] + ARGV[1]
			if qty < 0 then
			    return 'N:' .. stock[1]
			end
			local res = redis.call('HMSET', KEYS[1], 'quantity', qty, 'version', stock[2] + 1)
			if res.ok then
			    return 'Y:' .. stock[1] .. ';' .. stock[2]
			else
			    return 'N:ERR'
			end
			""", String.class);

	/**
	 * 初始化数据
	 */
	@PostConstruct
	void init() {
		LockInfo lockInfo = lockTemplate.lock("init_stock_cache", 5 * 60 * 1000, 500);
		if (lockInfo == null) {
			log.info("其它实例正在初始化库存缓存，当前实例跳过执行");
			return;
		}

		try {
			for (Stock stock : list()) {
				String key = getCacheKey(stock);
				Optional<Stock> optional = getCache(key);
				// 缓存数据
				if (optional.isEmpty()) {
					cacheStock(stock, key);
					continue;
				}
				// 更新缓存
				Stock cache = optional.get();
				long v = stock.getVersion() - cache.getVersion();
				if (v > 0) {
					cacheStock(stock, key);
					continue;
				}
				// 更新数据
				if (v < 0) {
					sendUpdateMessage(cache);
				}
				log.info("缓存成功：{}", key);
			}
			log.info("初始化库存缓存成功");
		} catch (Exception e) {
			log.error("初始化库存缓存失败：", e);
		} finally {
			lockTemplate.releaseLock(lockInfo);
		}
	}

	@Override
	public List<StockVO> list(IPage<Stock> page, Stock stock) {
		LambdaQueryWrapper<Stock> wrapper = Wrappers.lambdaQuery();
		wrapper.eq(Objects.nonNull(stock.getId()), Stock::getId, stock.getId());
		wrapper.eq(StrUtil.isNotBlank(stock.getWarehouseCode()), Stock::getWarehouseCode, stock.getWarehouseCode());
		wrapper.eq(StrUtil.isNotBlank(stock.getChannelCode()), Stock::getChannelCode, stock.getChannelCode());
		wrapper.eq(StrUtil.isNotBlank(stock.getSkuCode()), Stock::getSkuCode, stock.getSkuCode());
		wrapper.eq(StrUtil.isNotBlank(stock.getLocation()), Stock::getLocation, stock.getLocation());
		List<Stock> stocks = list(page, wrapper);
		List<StockVO> voList = new LinkedList<>();
		for (Stock s : stocks) {
			StockVO vo = StockConverter.INSTANCE.toStockVO(s);
			vo.setQuantityInfo(logService.quantityInfo(s.getId()));
			voList.add(vo);
		}
		return voList;
	}

	@Override
	public R adjust(StockDTO dto) {
		StockUpdateDTO su = StockUpdateDTO.builder().stockDTO(dto)
				.operation(StockConstants.Operation.ADJUST)
				.status(StockConstants.LogStatus.FINISH)
				.build();

		R res = validate(su);
		if (res.failure()) {
			return res;
		}
		return execute(su);
	}

	@Override
	public R batchAdjust(List<StockDTO> list, boolean rollback) {
		R res;
		StockUpdateDTO su = StockUpdateDTO.builder()
				.operation(StockConstants.Operation.BATCH_ADJUST)
				.status(StockConstants.LogStatus.FINISH)
				.build();

		// 数据验证
		List<String> errorList = new LinkedList<>();
		for (StockDTO dto : list) {
			su.setStockDTO(dto);
			res = validate(su);
			if (res.failure()) {
				errorList.add(res.getMsg());
			}
		}
		if (CollectionUtil.isNotEmpty(errorList)) {
			return R.failed(errorList);
		}

		// 批量调整
		List<StockDTO> successList = new LinkedList<>();
		for (StockDTO dto : list) {
			su.setStockDTO(dto);
			res = execute(su);
			if (res.failure()) {
				errorList.add(res.getMsg());
			} else {
				successList.add(dto);
			}
		}

		// 执行出错，回滚成功数据
		if (!rollback) {
			successList.clear();
		}
		for (StockDTO dto : successList) {
			dto.setQuantity(dto.getQuantity() * -1);
			su.setStockDTO(dto);
			res = execute(su);
			if (res.failure()) {
				log.error("批量调整回滚失败：{}\ndto:{}", res.getMsg(), JSON.toJSONString(dto, false));
			}
		}

		if (CollectionUtil.isEmpty(errorList)) {
			return R.ok();
		} else {
			return R.failed(errorList);
		}
	}

	@Override
	public R inTransit(StockDTO dto) {
		dto.setQuantity(Math.abs(dto.getQuantity()));
		StockUpdateDTO su = StockUpdateDTO.builder().stockDTO(dto)
				.operation(StockConstants.Operation.IN_TRANSIT)
				.update(false)
				.build();

		R res = validate(su);
		if (res.failure()) {
			return res;
		}
		return execute(su);
	}

	@Override
	public R inbound(StockDTO dto) {
		dto.setQuantity(Math.abs(dto.getQuantity()));
		StockUpdateDTO su = StockUpdateDTO.builder().stockDTO(dto)
				.operation(StockConstants.Operation.INBOUND)
				.status(StockConstants.LogStatus.FINISH)
				.build();

		R res = validate(su);
		if (res.failure()) {
			return res;
		}
		return execute(su);
	}

	@Override
	public R allocate(StockDTO dto) {
		dto.setQuantity(-1 * Math.abs(dto.getQuantity()));
		StockUpdateDTO su = StockUpdateDTO.builder().stockDTO(dto)
				.operation(StockConstants.Operation.ALLOCATE)
				.build();

		R res = validate(su);
		if (res.failure()) {
			return res;
		}
		return execute(su);
	}

	@Override
	public R outbound(StockDTO dto) {
		dto.setQuantity(-1 * Math.abs(dto.getQuantity()));
		StockUpdateDTO su = StockUpdateDTO.builder().stockDTO(dto)
				.operation(StockConstants.Operation.OUTBOUND)
				.update(false)
				.status(StockConstants.LogStatus.FINISH)
				.build();

		R res = validate(su);
		if (res.failure()) {
			return res;
		}
		return execute(su);
	}

	/**
	 * 数据验证
	 *
	 * @param su 操作数据
	 * @return
	 */
	R<String> validate(StockUpdateDTO su) {
		// 基础验证
		if (su == null) {
			return R.failedMessage(CommonErrorCodes.EMPTY_ERROR);
		}

		StockDTO dto = su.getStockDTO();
		if (dto == null) {
			return R.failedMessage(CommonErrorCodes.EMPTY_ERROR);
		}
		if (StrUtil.isBlank(dto.getWarehouseCode())) {
			return R.failedMessage(WsmsErrorCodes.WAREHOUSE_NOT_EXISTS, "");
		}
		if (StrUtil.isBlank(dto.getLocation())) {
			return R.failedMessage(WsmsErrorCodes.LOCATION_NOT_EXISTS, "");
		}
		if (StrUtil.isBlank(dto.getSkuCode())) {
			return R.failedMessage(WsmsErrorCodes.SKU_NOT_EXISTS, "");
		}
		if (dto.getQuantity() == null || dto.getQuantity() == 0) {
			return R.failedMessage(WsmsErrorCodes.QUANTITY_ILLEGAL);
		}

		// 仅完结日志状态的操作必须要有业务数据
		if (!su.isUpdate() && su.getStatus() == StockConstants.LogStatus.FINISH && StrUtil.isBlank(dto.getDataKey())) {
			return R.failedMessage(WsmsErrorCodes.DATA_KEY_EMPTY);
		}

		// 默认渠道编码
		if (StrUtil.isBlank(dto.getChannelCode())) {
			dto.setChannelCode(StockConstants.DEFAULT_CHANNEL);
		} else {
			if (channelService.getById(dto.getChannelCode()) == null) {
				return R.failedMessage(WsmsErrorCodes.CHANNEL_NOT_EXISTS, dto.getChannelCode());
			}
		}

		// 数据验证
		if (warehouseService.getById(dto.getWarehouseCode()) == null) {
			return R.failedMessage(WsmsErrorCodes.WAREHOUSE_NOT_EXISTS, dto.getWarehouseCode());
		}
		if (!StockLocationEnum.validate(dto.getLocation())) {
			return R.failedMessage(WsmsErrorCodes.LOCATION_NOT_EXISTS, dto.getLocation());
		}
		if (skuService.getById(dto.getSkuCode()) == null) {
			return R.failedMessage(WsmsErrorCodes.SKU_NOT_EXISTS, dto.getSkuCode());
		}

		return R.ok();
	}

	/**
	 * 执行库存操作
	 *
	 * @param su 操作数据
	 * @return
	 */
	R execute(StockUpdateDTO su) {
		R res;

		// 获取库存记录
		StockDTO dto = su.getStockDTO();
		Optional<Stock> optional = getCache(getCacheKey(dto));
		Stock stock;
		if (optional.isEmpty()) {
			// 出库时，不允许操作不存在的库存
			if (dto.getQuantity() < 0) {
				return R.failedMessage(WsmsErrorCodes.STOCK_NOT_EXISTS, dto.getWarehouseCode(), dto.getSkuCode());
			}

			// 创建空库存
			res = createInitStock(su);
			if (res.failure()) {
				return res;
			}
			stock = (Stock) res.getData();
		} else {
			stock = optional.get();
		}
		su.setStock(stock);

		// 更新库存缓存
		res = updateCache(su);
		if (res.failure()) {
			return res;
		}

		// 构建库存日志
		StockLog stockLog = logService.build(su);

		// 更新库存数据
		if (su.isUpdate()) {
			stock.setQuantity(stock.getQuantity() + dto.getQuantity());
			stock.setVersion(stock.getVersion() + 1);
			sendUpdateMessage(stock);
		}

		// 记录库存日志
		logService.asyncLog(stockLog);

		return R.ok(stock);
	}

	/**
	 * 创建初始库存数据
	 *
	 * @param su 库存操作数据
	 * @return 执行结果和库存记录
	 */
	R createInitStock(StockUpdateDTO su) {
		R res;
		StockDTO dto = su.getStockDTO();
		String key = getCacheKey(dto);
		LockInfo lockInfo = lockTemplate.lock(RedisUtils.getLockKey(key));
		try {
			// 再次获取避免并发问题
			Optional<Stock> stockOptional = getCache(key);
			if (stockOptional.isPresent()) {
				return R.ok(stockOptional.get());
			}

			// 保存到数据库
			Stock stock = new Stock();
			stock.setWarehouseCode(dto.getWarehouseCode());
			stock.setChannelCode(dto.getChannelCode());
			stock.setSkuCode(dto.getSkuCode());
			stock.setLocation(dto.getLocation());
			boolean result = save(stock);
			if (!result) {
				return R.failedMessage(CommonErrorCodes.SAVE_ERROR);
			}

			cacheStock(stock, key);

			res = R.ok(stock);
		} catch (Exception e) {
			log.error("库存创建失败：", e);
			res = R.failedMessage(CommonErrorCodes.ERROR, ExceptionUtil.getSimpleMessage(e));
		} finally {
			lockTemplate.releaseLock(lockInfo);
		}

		return res;
	}

	/**
	 * 获取缓存键名称
	 *
	 * @param obj StockDTO/Stock类型对象
	 * @return
	 */
	String getCacheKey(Object obj) {
		String warehouseCode = "";
		String channelCode = "";
		String skuCode = "";
		String location = "";

		if (obj instanceof StockDTO dto) {
			warehouseCode = dto.getWarehouseCode();
			channelCode = dto.getChannelCode();
			skuCode = dto.getSkuCode();
			location = dto.getLocation();
		}

		if (obj instanceof Stock stock) {
			warehouseCode = stock.getWarehouseCode();
			channelCode = stock.getChannelCode();
			skuCode = stock.getSkuCode();
			location = stock.getLocation();
		}

		return RedisUtils.getCacheKey("Stock", warehouseCode, channelCode, skuCode, location);
	}

	/**
	 * 创建缓存数据
	 *
	 * @param stock
	 * @param key   选填
	 */
	void cacheStock(Stock stock, String key) {
		if (StrUtil.isBlank(key)) {
			key = getCacheKey(stock);
		}
		stringRedisTemplate.opsForHash().putAll(key, StockConverter.INSTANCE.toStringMap(stock));
	}

	/**
	 * 更新缓存数据
	 *
	 * @param su
	 * @return
	 */
	R updateCache(StockUpdateDTO su) {
		Stock stock = su.getStock();
		try {
			// 无需更新缓存
			if (!su.isUpdate()) {
				return R.ok(stock);
			}

			// 更新库存数据
			StockDTO dto = su.getStockDTO();
			List<String> keys = Collections.singletonList(getCacheKey(dto));
			String result = stringRedisTemplate.execute(STOCK_SCRIPT, keys, dto.getQuantity().toString());

			// 解析返回的执行结果和库存数据
			if (StrUtil.isBlank(result) || StrUtil.equals(result, "FAIL")) {
				return R.failedMessage(WsmsErrorCodes.STOCK_UPDATE_FAILED, stock.getId().toString());
			}

			String[] data = result.split(":");
			if (StrUtil.equals(data[0], "N")) {
				if (StrUtil.equals(data[1], "ERR")) {
					return R.failedMessage(WsmsErrorCodes.STOCK_UPDATE_FAILED, stock.getId().toString());
				} else {
					return R.failedMessage(WsmsErrorCodes.QUANTITY_NOT_ENOUGH, stock.getId().toString(), data[1]);
				}
			}

			data = data[1].split(";");
			if (StrUtil.isBlank(data[0])) {
				stock.setQuantity(Integer.parseInt(data[0]));
			}
			if (StrUtil.isBlank(data[1])) {
				stock.setVersion(Long.parseLong(data[1]));
			}
		} catch (Exception e) {
			log.error("库存缓存{}更新失败：", stock.getId(), e);
			return R.failedMessage(WsmsErrorCodes.STOCK_UPDATE_FAILED, stock.getId().toString());
		}

		return R.ok(stock);
	}

	/**
	 * 获取缓存数据
	 *
	 * @param key 库存缓存键
	 * @return 库存记录
	 */
	Optional<Stock> getCache(String key) {
		HashOperations<String, String, String> opsForHash = stringRedisTemplate.opsForHash();
		Map<String, String> cache = opsForHash.entries(key);
		if (CollectionUtil.isEmpty(cache)) {
			return Optional.empty();
		} else {
			return Optional.of(StockConverter.INSTANCE.toStock(cache));
		}
	}

	/**
	 * 发送数据更新消息
	 *
	 * @param stock
	 * @return
	 */
	void sendUpdateMessage(Stock stock) {
		try {
			if (stock == null) {
				return;
			}
			amqpTemplate.convertAndSend(WsmsQueues.STOCK_UPDATE, stock);
		} catch (Exception e) {
			log.error("库存记录{}请求更新失败：", JSON.toJSONString(stock, false), e);
		}
	}

	/**
	 * 监听数据更新队列
	 */
	@RabbitListener(queuesToDeclare = @Queue(WsmsQueues.STOCK_UPDATE))
	void onUpdateMessage(Stock stock) {
		try {
			LambdaUpdateWrapper<Stock> wrapper = Wrappers.lambdaUpdate();
			wrapper.set(Stock::getQuantity, stock.getQuantity());
			wrapper.set(Stock::getVersion, stock.getVersion());
			wrapper.eq(Stock::getId, stock.getId());
			wrapper.lt(Stock::getVersion, stock.getVersion());
			update(wrapper);
			log.info("库存记录更新成功：ID:{};version:{}", stock.getId(), stock.getVersion());
		} catch (Exception e) {
			log.error("库存记录{}更新失败：", JSON.toJSONString(stock, false), e);
		}
	}

}